
package be.ibridge.kettle.kgeocode;

import java.net.HttpURLConnection;
//import java.net.MalformedURLException;
import java.net.URL;
import java.net.URLEncoder;

import javax.xml.parsers.DocumentBuilderFactory;
import javax.xml.xpath.XPath;
import javax.xml.xpath.XPathConstants;
import javax.xml.xpath.XPathFactory;

import org.w3c.dom.Document; 
import org.w3c.dom.Node; 
import org.w3c.dom.NodeList; 
import org.xml.sax.InputSource; 
 
import java.io.IOException; 
//import java.io.UnsupportedEncodingException;

import javax.xml.parsers.ParserConfigurationException; 
import javax.xml.xpath.XPathExpressionException; 
import org.xml.sax.SAXException;

import org.pentaho.di.core.Const;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.row.RowDataUtil;
import org.pentaho.di.core.row.RowMetaInterface;
import org.pentaho.di.core.row.ValueMeta;
import org.pentaho.di.core.row.ValueMetaInterface;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.BaseStep;
import org.pentaho.di.trans.step.StepDataInterface;
import org.pentaho.di.trans.step.StepInterface;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.step.StepMetaInterface;

public class KGeoCodingPlugin extends BaseStep implements StepInterface
{
    private KGeoCodingPluginData data;
	private KGeoCodingPluginMeta meta;
	private int ColPos; // adress field position
    Object[] outputRow;
	private static final String GEOCODER_REQUEST_PREFIX_FOR_XML = "http://maps.google.com/maps/api/geocode/xml";
	
	public KGeoCodingPlugin(StepMeta s, StepDataInterface stepDataInterface, int c, TransMeta t, Trans dis)
	{
		super(s,stepDataInterface,c,t,dis);     
	}
	
	public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException
	{
	    meta = (KGeoCodingPluginMeta)smi;
	    data = (KGeoCodingPluginData)sdi;
	    
		Object[] r=getRow();    // get row, blocks when needed!
		if (r==null)  // no more input to be expected...
		{
			setOutputDone();
			return false;
		}
        
        if (first)
        {
            first = false;
            
            data.firstNewFieldIndex = getInputRowMeta().size();
            data.outputRowMeta = (RowMetaInterface)getInputRowMeta().clone();
            meta.getFields(data.outputRowMeta, getStepname(), null, null, this);  
            /*
            if (!meta.getNormalizedAddress().isEmpty()){
        	ValueMetaInterface AddressMeta=new ValueMeta(meta.getNormalizedAddress().toString(), ValueMetaInterface.TYPE_STRING);
        	data.outputRowMeta.addValueMeta(AddressMeta); 
            }
            if (!meta.getCity().isEmpty()){
        	ValueMetaInterface CityMeta=new ValueMeta(meta.getCity().toString(), ValueMetaInterface.TYPE_STRING);
        	data.outputRowMeta.addValueMeta(CityMeta);
            }
            if (!meta.getGPSCoord().isEmpty()){
        	ValueMetaInterface GPSMeta=new ValueMeta(meta.getGPSCoord().toString(), ValueMetaInterface.TYPE_STRING);
        	data.outputRowMeta.addValueMeta(GPSMeta);
            }
            */
    	    
        }
        String retour = "";
        
        try {
        	
        	int tt = data.firstNewFieldIndex;
        	logBasic ("Valeur de firstfieldindex : " + tt);
        	ColPos = data.outputRowMeta.indexOfValue(meta.getAddressField());
        	
        	
			retour = GeoCodeThis(r[ColPos].toString());
			logBasic("Retour Geocoding : " + retour);
			String[] items;
		    items = retour.split("#");
		     //reserve room
		    Object[] outputRow = RowDataUtil.allocateRowData(data.outputRowMeta.size());
				for (int i = 0; i < data.firstNewFieldIndex; i++)
				{
					outputRow[i] = r[i];
				}

				outputRow[data.firstNewFieldIndex]   = items[0]; // Normalized address

				outputRow[data.firstNewFieldIndex+1] = items[1]; // City

				outputRow[data.firstNewFieldIndex+2] = items[2]; // GPS coord
		        
				putRow(data.outputRowMeta, outputRow);     // copy row to possible alternate rowset(s).
			
			
		} catch (XPathExpressionException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (SAXException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (ParserConfigurationException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}  
        
		if (checkFeedback(linesRead)) logBasic("Linenr "+linesRead);  // Some basic logging every 5000 rows.
		return true;
	}
	
	public String GeoCodeThis(String InputAddress) throws IOException, SAXException, ParserConfigurationException, XPathExpressionException{
		 String NormAdd ="";
	     String Locality = "";
	     String GPSC = "";
	  
	     // prepare a URL to the geocoder 
	     URL url = new URL(GEOCODER_REQUEST_PREFIX_FOR_XML + "?address=" + URLEncoder.encode(InputAddress, "UTF-8") + "&sensor=false"); 
	  
	     HttpURLConnection conn = (HttpURLConnection) url.openConnection(); 
	  
	     Document geocoderResultDocument = null; 
	     try { 
	       conn.connect(); 
	       InputSource geocoderResultInputSource = new InputSource(conn.getInputStream()); 
	       geocoderResultDocument = DocumentBuilderFactory.newInstance().newDocumentBuilder().parse(geocoderResultInputSource); 
	     } finally { 
	       conn.disconnect(); 
	     } 
	     XPath xpath = XPathFactory.newInstance().newXPath(); 

	     NodeList resultNodeList = null; 

	     resultNodeList = (NodeList) xpath.evaluate("/GeocodeResponse/result/formatted_address", geocoderResultDocument, XPathConstants.NODESET); 
	     for(int i=0; i<resultNodeList.getLength(); ++i) { 
	       NormAdd += resultNodeList.item(i).getTextContent();
	     } 
	     
	     resultNodeList = (NodeList) xpath.evaluate("/GeocodeResponse/result[1]/address_component[type/text()='locality']/long_name", geocoderResultDocument, XPathConstants.NODESET); 
	     for(int i=0; i<resultNodeList.getLength(); ++i) { 
	       Locality += resultNodeList.item(i).getTextContent();
	     } 

	     resultNodeList = (NodeList) xpath.evaluate("/GeocodeResponse/result[1]/geometry/location/*", geocoderResultDocument, XPathConstants.NODESET); 
	     float lat = Float.NaN; 
	     float lng = Float.NaN; 
	     for(int i=0; i<resultNodeList.getLength(); ++i) { 
	       Node node = resultNodeList.item(i); 
	       if("lat".equals(node.getNodeName())) lat = Float.parseFloat(node.getTextContent()); 
	       if("lng".equals(node.getNodeName())) lng = Float.parseFloat(node.getTextContent()); 
	     } 
	     GPSC = lat + "//" + lng;
		return NormAdd + "#" + Locality + "#" + GPSC ; 
	   } 
	
	

	public boolean init(StepMetaInterface smi, StepDataInterface sdi)
	{
	    meta = (KGeoCodingPluginMeta)smi;
	    data = (KGeoCodingPluginData)sdi;

	    return super.init(smi, sdi);
	}

	public void dispose(StepMetaInterface smi, StepDataInterface sdi)
	{
	    meta = (KGeoCodingPluginMeta)smi;
	    data = (KGeoCodingPluginData)sdi;

	    super.dispose(smi, sdi);
	}
	
	//
	// Run is were the action happens!
	public void run()
	{
		logBasic("Starting to run...");
		try
		{
			while (processRow(meta, data) && !isStopped());
		}
		catch(Exception e)
		{
			logError("Unexpected error : "+e.toString());
            logError(Const.getStackTracker(e));
			setErrors(1);
			stopAll();
		}
		finally
		{
		    dispose(meta, data);
			logBasic("Finished, processing "+linesRead+" rows");
			markStop();
		}
	}
}
